package com.soft863.tc.analysis.io;

import com.soft863.tc.common.constant.ValueConstant;
import com.soft863.tc.common.util.ConnectionUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.PathOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.PathOutputCommitterFactory;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.*;

/*
 * mysql数据格式化输出对象
 * */
public class MySQLTextOutputFormat extends OutputFormat<Text, Text> {
    private PathOutputCommitter committer = null;

    protected static class MySQLRecordWriter extends RecordWriter<Text, Text> {

        @Override
        public void write(Text key, Text value) throws IOException, InterruptedException {
            //133_20180909
            String[] kValues = key.toString().split("_");
            String[] vValues = value.toString().split("_");
            PreparedStatement ps = null;
            try {
                String sql = "insert into tc_call(tel_id, date_id, sum_call, sum_duration) values(?,?,?,?)";
                ps = ConnectionUtil.getMySQLConnection().prepareStatement(sql);

                //-1 表示没找到缓存数据
                Jedis jedis = ConnectionUtil.getJedis();
                String telId = jedis.hget(ValueConstant.REDIS_HASH_USER, kValues[0]);
                if (telId == null) {
                    telId = "-1";
                }
                String dateId = jedis.hget(ValueConstant.REDIS_HASH_DATE, kValues[1]);
                if (dateId == null) {
                    dateId = "-1";
                }
                ps.setInt(1, Integer.parseInt(telId));
                ps.setInt(2, Integer.parseInt(dateId));
                ps.setInt(3, Integer.parseInt(vValues[0]));
                ps.setInt(4, Integer.parseInt(vValues[1]));
                ps.executeUpdate();
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                if (ps != null) {
                    try {
                        ps.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            ConnectionUtil.closeMySQL();
        }
    }

    @Override
    public RecordWriter<Text, Text> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
        return new MySQLRecordWriter();
    }

    @Override
    public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException {

    }

    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException {
        if (committer == null) {
            Path output = getOutputPath(context);
            committer = PathOutputCommitterFactory.getCommitterFactory(
                    output,
                    context.getConfiguration()).createOutputCommitter(output, context);
        }
        return committer;
    }

    public static Path getOutputPath(JobContext job) {
        String name = job.getConfiguration().get(FileOutputFormat.OUTDIR);
        return name == null ? null : new Path(name);
    }
}
